package com.yht.redbook.data.align.consumer;

import com.yht.framework.common.utils.JsonUtils;
import com.yht.redbook.data.align.constant.MQConstants;
import com.yht.redbook.data.align.constant.RedisKeyConstants;
import com.yht.redbook.data.align.constant.TableConstants;
import com.yht.redbook.data.align.domain.mapper.InsertRecordMapper;
import com.yht.redbook.data.align.model.dto.NoteOperateMqDTO;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.data.redis.core.script.RedisScript;
import org.springframework.scripting.support.ResourceScriptSource;
import org.springframework.stereotype.Component;

import java.time.LocalDate;
import java.time.format.DateTimeFormatter;
import java.util.Collections;
import java.util.Objects;

@RocketMQMessageListener(consumerGroup = "redbook_group_data_align_" + MQConstants.TOPIC_NOTE_OPERATE,
        topic = MQConstants.TOPIC_NOTE_OPERATE)
@Component
@Slf4j
public class TodayNotePublishIncrementData2DBConsumer implements RocketMQListener<String> {

    @Resource
    private RedisTemplate<String, Object> redisTemplate;

    @Resource
    private InsertRecordMapper insertRecordMapper;

    @Value("${table.shards}")
    private int tableShards;

    @Override
    public void onMessage(String body) {
        log.info("## TodayNotePublishIncrementData2DBConsumer 消费到了 MQ: {}", body);

        // 布隆过滤器判断该日增量数据是否已经记录
        NoteOperateMqDTO noteOperateMqDTO = JsonUtils.parseObject(body, NoteOperateMqDTO.class);
        if (Objects.isNull(noteOperateMqDTO)) return;

        // 发布删除 ID
        Long creatorId = noteOperateMqDTO.getCreatorId();

        String date = LocalDate.now().format(DateTimeFormatter.ofPattern("yyyyMMdd"));
        String bloomUserNoteOperateListKey = RedisKeyConstants.buildBloomUserNoteOperateListKey(date);

        DefaultRedisScript<Long> script = new DefaultRedisScript<>();
        script.setScriptSource(new ResourceScriptSource(new ClassPathResource("lua/bloom_today_user_note_publish_check.lua")));
        script.setResultType(Long.class);

        Long result = redisTemplate.execute(script, Collections.singletonList(bloomUserNoteOperateListKey), creatorId);
        if (Objects.equals(result, 0L)) {
            // 若无，才会落库，减轻数据库压力
            // 将日增量变更数据，写入日增量表中
            // - t_data_align_note_publish_count_temp_日期_分片序号
            // 根据分片总数，取模，分别获取对应的分片序号
            long useIdHashKey = creatorId % tableShards;

            // 将日增量变更数据写入到日增量表中
            insertRecordMapper.insert2DataAlignUserNotePublishCountTemplate(TableConstants.buildTableNameSuffix(date, useIdHashKey), creatorId);

            // 3.数据库写入成功后，再添加到布隆过滤器中
            RedisScript<Long> bloomAddScript = RedisScript.of("return redis.call('BF.ADD', KEYS[1], ARGV[1])", Long.class);
            redisTemplate.execute(bloomAddScript, Collections.singletonList(bloomUserNoteOperateListKey), creatorId);
        }
    }
}
